package com.atguigu.edu.app.dwd.db;

import com.alibaba.druid.util.MySqlUtils;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import util.MyKafkaUtil;

//互动域评价事务事实表
public class DwdInteractionComment {
    public static void main(String[] args) {
        //TODO 1.环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //  1.1 设置全局并行度
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //TODO 2.检查点相关设置
        //TODO 3.从kafka中读取
        tableEnv.executeSql(MyKafkaUtil.getTopicDDL("dwd_interaction_comment"));
//        tableEnv.executeSql("select * from ODS_BASE_DB").print();
        // TODO 4.读取评论表数据
        Table reviewInfo = tableEnv.sqlQuery("select\n" +
                "data['id'] id,\n" +
                "data['user_id'] user_id,\n" +
                "data['course_id'] course_id,\n" +
                "data['review_txt'] review_txt,\n" +
                "data['review_stars'] review_stars,\n" +
                "data['create_time'] create_time,\n" +
                "data['deleted'] deleted,\n" +
                "proc_time,\n" +
                "ts\n" +
                "from ODS_BASE_DB\n" +
                "where `table` = 'review_info'\n" +
                "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("review_info",reviewInfo);
        //TODO 5.查询reviewInfo数据
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "id,\n" +
                "user_id,\n" +
                "course_id,\n" +
                "review_txt,\n" +
                "review_stars,\n" +
                "create_time,\n" +
                "deleted,\n" +
                "proc_time,\n" +
                "ts\n" +
                "from review_info \n");
        //TODO 6.建立Upsert-Kafka dwd_interaction_comment 表
        tableEnv.executeSql("create table dwd_interaction_comment(\n" +
                "id string,\n" +
                "user_id string,\n" +
                "course_id string,\n" +
                "review_txt string,\n" +
                "review_stars string,\n" +
                "create_time string,\n" +
                "deleted string,\n" +
                "proc_time timestamp_ltz(3),\n" +
                "ts string,\n" +
                "primary key(id) not enforced\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_interaction_comment"));

        //TODO 7.将关联结果写入Upsert-Kafka表
        tableEnv.executeSql("" +
                "insert into dwd_interaction_comment select * from review_info");
    }
}
